package com.atguigu.champter5.Flink05.transformation.UDF;


import com.atguigu.beans.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;

/**
 * @author Hefei
 * @description
 * @project_name flink
 * @package_name com.atguigu.chapter02
 * @since 2021/11/7 17:07
 */
public class TransFunctionLambda {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 模拟数据
        ArrayList<WaterSensor> waterSensors = new ArrayList<>();
        waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
        waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
        DataStreamSource<WaterSensor> ws = env.fromCollection(waterSensors);

        // 对数据使用map算子进行转换，但是转换之后的数据类型不包含泛型结构
        // 但如果是类似在第二章中的需求，需要通过map算子
        // 将word转换为Tuple2<String,Integer>这样的结构，则仍然需要显式指定返回值类型
        DataStream<String> stream = ws.map(waterSensor -> waterSensor.toString());

        /*

        DataStream<String> stream = ws.map(new MapFunction<WaterSensor, String>() {
            @Override
            public String map(WaterSensor value) throws Exception {
                return value.toString();
            }
        });

        */

        stream.print();
        env.execute();
    }
}
